package net.wicp.tams.common.binlog.alone.parser;

import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.Validate;
import org.mvel2.sh.Main;

import io.thekraken.grok.api.Match;
import io.thekraken.grok.api.exception.GrokException;
import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Conf;
import net.wicp.tams.common.apiext.GrokObj;
import net.wicp.tams.common.apiext.PwdUtil;
import net.wicp.tams.common.apiext.ReflectAssist;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.apiext.UUIDGenerator;
import net.wicp.tams.common.apiext.jdbc.JdbcAssit;
import net.wicp.tams.common.apiext.jdbc.JdbcConnection;
import net.wicp.tams.common.binlog.alone.ListenerConf.ColHis;
import net.wicp.tams.common.binlog.alone.ListenerConf.ColHis.Builder;
import net.wicp.tams.common.binlog.alone.ListenerConf.ConnConf;
import net.wicp.tams.common.binlog.alone.normalize.IBinlogListener;
import net.wicp.tams.common.binlog.alone.normalize.ISaveCheckPoint;
import net.wicp.tams.common.binlog.parser.LogEvent;
import net.wicp.tams.common.binlog.parser.event.GtidLogEvent;
import net.wicp.tams.common.binlog.parser.event.QueryLogEvent;
import net.wicp.tams.common.binlog.parser.event.RowsLogBuffer;
import net.wicp.tams.common.binlog.parser.event.RowsLogEvent;
import net.wicp.tams.common.binlog.parser.event.TableMapLogEvent;
import net.wicp.tams.common.binlog.parser.event.TableMapLogEvent.ColumnInfo;
import net.wicp.tams.common.binlog.parser.event.XidLogEvent;
import net.wicp.tams.common.constant.DateFormatCase;
import net.wicp.tams.common.constant.OptType;
import net.wicp.tams.common.constant.StrPattern;
import net.wicp.tams.common.constant.dic.YesOrNo;
import net.wicp.tams.common.exception.ExceptAll;
import net.wicp.tams.common.exception.ProjectExceptionRuntime;
import net.wicp.tams.duckula.client.Protobuf3.DuckulaEvent;

@Slf4j
public abstract class BaseLogFetcher {
	protected String fileName = "mysql-bin.000001";
	protected Charset charset = Charset.forName("utf-8");
	protected String gtids;

	private GrokObj gm = GrokObj.getInstance();
	{
		gm.addPattern("tb", "[A-Za-z0-9_.-:]+");
		gm.addPattern("tball", "alter\\s+table\\s+`?%{tb}`?");
		//
		gm.addPattern("db", "[A-Za-z0-9_.-:]+");
		gm.addPattern("tball2", "alter\\s+table\\s+`?%{db}`.`%{tb}`?");
	}

	protected final ConnConf connConf;
	protected final BinlogMetricGroup metric;
	private final IBinlogListener binlogListener;// 业务处理逻辑

	public IBinlogListener getBinlogListener() {
		return binlogListener;
	}

	protected ISaveCheckPoint saveCheckPoint;
	// 存放cols的map
	protected final Map<String, List<ColHis>> colsMap = new HashMap<String, List<ColHis>>();

	protected BaseLogFetcher(ConnConf.Builder connConfBuilder) {
		Validate.notBlank(connConfBuilder.getIp(), "需要ip地址");
		Validate.notBlank(connConfBuilder.getUser(), "需要用户名");
		Validate.notBlank(connConfBuilder.getPassword(), "需要密码");
		Validate.notBlank(connConfBuilder.getListener(), "需要处理的监听类");
		if (connConfBuilder.getPort() <= 0) {
			connConfBuilder.setPort(3306);
		}
		if (connConfBuilder.getClientId() <= 0) {
			connConfBuilder.setClientId(StringUtil.buildPort(new UUIDGenerator().generate()));
		}
		if (StringUtil.isNull(connConfBuilder.getDbPattern())) {
			connConfBuilder.setDbPattern("^.*$");
		}
		if (StringUtil.isNull(connConfBuilder.getTbPattern())) {
			connConfBuilder.setTbPattern("^.*$");
		}
		if (StringUtil.isNull(connConfBuilder.getChk())) {
			connConfBuilder.setChk("net.wicp.tams.common.binlog.alone.checkpoint.CheckPointH2db");
		}
		// net.wicp.tams.common.binlog.alone.checkpoint.CheckPointH2db
		try {
			saveCheckPoint = (ISaveCheckPoint) Class.forName(connConfBuilder.getChk()).newInstance();
			saveCheckPoint.init(connConfBuilder);
		} catch (Exception e1) {
			throw new ProjectExceptionRuntime(ExceptAll.project_other, "创建checkpoint实例失败");
		}
		// 需要依赖saveCheckPoint
		  init(connConfBuilder);
		// 先获得分布式锁
		YesOrNo acquireLock = saveCheckPoint.acquireLock();
		if (acquireLock == null || acquireLock == YesOrNo.no) {
			throw new ProjectExceptionRuntime(ExceptAll.project_other, "不能获得分布式锁");
		}
		
		this.connConf = connConfBuilder.build();
		this.metric = new BinlogMetricGroup(String.format("%s:%s", connConf.getIp(), connConf.getPort()));
		Class<?> class1 = null;
		try {
			class1 = Class.forName(connConf.getListener());
		} catch (ClassNotFoundException e1) {
			log.error("没有指定的Listener");
			throw new ProjectExceptionRuntime(ExceptAll.Param_typenofit, "没有指定的Listener");
		}
		if (!ReflectAssist.isInterface(class1, "net.wicp.tams.common.binlog.alone.normalize.IBinlogListener")) {
			log.error("Listener需要net.wicp.tams.common.binlog.alone.IBinlogListener类型");
			throw new ProjectExceptionRuntime(ExceptAll.Param_typenofit,
					"Listener需要net.wicp.tams.common.binlog.alone.IBinlogListener类型");
		}
		try {
			this.binlogListener = (IBinlogListener) ReflectAssist.newInst(class1);
			this.binlogListener.init();
		} catch (Exception e1) {
			log.error("Listener实例化失败");
			throw new ProjectExceptionRuntime(ExceptAll.Param_typenofit, "Listener实例化失败");
		}
	}

	protected void parseQueryEvent(QueryLogEvent event) {
		String db = event.getDbName();
		String sql = event.getQuery().toLowerCase();
		if (sql.startsWith("alter")) {
			try {
				Match match = gm.match("%{tball}", sql);
				String tb = String.valueOf(match.toMap().get("tb"));
				tb = tb.replace(db + ".", "");// 有可能带db进行修改`binlog_test_db.user_info`
				if (db.equals(tb)) {
					// 适合"alter table `binlog_test_db`.`user_info` "情况
					Match match2 = gm.match("%{tball2}", sql);
					tb = String.valueOf(match2.toMap().get("tb"));
				}
				if (!isValid(db, tb)) {
					return;
				}
				findCols(db, tb, event.getWhen());
				/*
				 * String key = String.format("%s|%s", db, tb).toLowerCase(); if
				 * (colsMap.containsKey(key)) { colsMap.get(key).add(findCols);
				 * log.warn("db:{},tb:{},cols is altered", db, tb); } else { List<ColHis>
				 * templist = new ArrayList<ColHis>(); templist.add(findCols);// 最开始那个col为默认值
				 * colsMap.put(key, templist); log.info("db:{},tb:{},cols add in map", db, tb);
				 * }
				 */
			} catch (GrokException e) {
				log.error("get tb from sql error", e);
			}
		}
	}

	protected void parseGtidLogEvent(GtidLogEvent event) throws Exception {
		this.gtids = event.getGtid();
		parseGtidLogEventSub(event);
	}

	protected abstract void parseGtidLogEventSub(GtidLogEvent event);

	/****
	 * 初始化一些变量
	 * 
	 * @param connConfBuilder
	 */
	protected abstract void init(ConnConf.Builder connConfBuilder);

	// 关闭相关资源
	public abstract void close();

	/***
	 * 开始读binlog
	 */
	public abstract void read();

	protected long xid;

	// gtid事件结束
	protected void parseXidEvent(XidLogEvent event) {
		this.xid = event.getXid();
	}

	protected boolean isChkDb=false;
	private boolean isValid(String db, String tb) {
		if (db.equals(Conf.get("common.binlog.alone.chk.mysql.defaultdb"))) {// 去除mysql数据库"^.*$".equals(connConf.getDbPattern())，它会存在
																				// ha_health_check 心跳表
			isChkDb=true;
			return false;
		}else {
			isChkDb=false;
		}
		return StrPattern.checkStrFormat(connConf.getDbPattern(), db)
				&& StrPattern.checkStrFormat(connConf.getTbPattern(), tb);
	}

	protected boolean parseRowsEvent(RowsLogEvent event, OptType optType) {
		if (log.isDebugEnabled()) {
			Date d = new Date(event.getHeader().getWhen() * 1000);
			String datestr = DateFormatCase.YYYY_MM_DD_hhmmss.getInstanc().format(d);
			log.debug("db:{},tb:{},time:{}", event.getTable().getDbName(), event.getTable().getTableName(), datestr);
		}
		if (!isValid(event.getTable().getDbName(), event.getTable().getTableName())) {// 没有匹配规则，直接跳过
			return false;
		}
		metric.meter_parser_pack_row.mark();
		ColHis colhis = ValidKey(event.getTable().getDbName(), event.getTable().getTableName(), event.getWhen());
		List<String[]> rowListBefore = new ArrayList<>();
		List<String[]> rowListAfter = new ArrayList<>();
		int rows = 0;
		try {
			RowsLogBuffer buffer = event.getRowsBuf(charset.name());
			BitSet columns = event.getColumns();
			BitSet changeColumns = event.getChangeColumns();
			while (buffer.nextOneRow(columns)) {
				// 处理row记录
				int type = event.getHeader().getType();
				if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) {
					// insert的记录放在before字段中
					rowListAfter.add(parseOneRow(event, buffer, columns, colhis.getColsCount(), true));
				} else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type) {
					// delete的记录放在before字段中
					rowListBefore.add(parseOneRow(event, buffer, columns, colhis.getColsCount(), false));
				} else {
					// update需要处理before/after
					// System.out.println("-------> before");
					rowListBefore.add(parseOneRow(event, buffer, columns, colhis.getColsCount(), true));
					if (!buffer.nextOneRow(changeColumns)) {
						break;
					}
					// System.out.println("-------> after");
					rowListAfter.add(parseOneRow(event, buffer, changeColumns, colhis.getColsCount(), true));
				}
				rows++;
			}
		} catch (Exception e) {
			throw new RuntimeException("parse row data failed.", e);
		}

		String[][] afterArray = new String[rowListAfter.size()][];
		for (int i = 0; i < rowListAfter.size(); i++) {
			afterArray[i] = rowListAfter.get(i);
		}

		String[][] beforeArray = new String[rowListBefore.size()][];
		for (int i = 0; i < rowListBefore.size(); i++) {
			beforeArray[i] = rowListBefore.get(i);
		}

		DuckulaEvent.Builder duckulaEventBuilder = DuckulaEvent.newBuilder();
		duckulaEventBuilder.setColNum(ArrayUtils.isNotEmpty(afterArray) ? afterArray[0].length : beforeArray[0].length);

		duckulaEventBuilder.addAllCols(colhis.getColsList());
		ColumnInfo[] colsColumnInfo = event.getTable().getColumnInfo();
		for (int i = 0; i < colsColumnInfo.length; i++) {
			duckulaEventBuilder.addColsTypeValue(colsColumnInfo[i].type);
		}
		duckulaEventBuilder.setDb(event.getTable().getDbName());
		duckulaEventBuilder.setTb(event.getTable().getTableName());
		duckulaEventBuilder.setGtid(this.gtids);
		duckulaEventBuilder.setIsError(false);
		duckulaEventBuilder.setOptType(net.wicp.tams.duckula.client.Protobuf3.OptType.forNumber(optType.getValue()));

		for (int i = 0; i < rows; i++) {
			DuckulaEvent.Builder rowBuilder = duckulaEventBuilder.clone();
			initData(rowBuilder, rowListBefore, colhis.getColsList(), i);
			initData(rowBuilder, rowListAfter, colhis.getColsList(), i);
			this.binlogListener.doBui(rowBuilder.build());
		}
		metric.meter_parser_event.mark(rows);
		metric.counter_ringbuff_pack.inc();// 包数
		metric.counter_ringbuff_event.inc(rows);// 记录数
		return true;
	}

	protected String[] parseOneRow(RowsLogEvent event, RowsLogBuffer buffer, BitSet cols, int colsNum, boolean isAfter)
			throws UnsupportedEncodingException {
		TableMapLogEvent map = event.getTable();
		if (map == null) {
			throw new RuntimeException("not found TableMap with tid=" + event.getTableId());
		}
		if (colsNum != event.getTable().getColumnCnt()) {
			throw new RuntimeException("TableMap:" + event.getTableId() + " the colsname is:" + colsNum
					+ " the value size is:" + event.getTable().getColumnCnt());
		}
		String[] values = new String[event.getTable().getColumnCnt()];
		final int columnCnt = map.getColumnCnt();
		final ColumnInfo[] columnInfo = map.getColumnInfo();
		for (int i = 0; i < columnCnt; i++) {
			if (!cols.get(i)) {
				continue;
			}
			ColumnInfo info = columnInfo[i];
			buffer.nextValue(info.type, info.meta);

			if (buffer.isNull()) {
				//
			} else {
				final Serializable value = buffer.getValue();
				if (value instanceof byte[]) {
					values[i] = PwdUtil.base64FromBin((byte[]) value);// new
																		// String((byte[])
																		// value);
				} else {
					values[i] = String.valueOf(value);
				}
			}
		}
		return values;
	}

	private PreparedStatement prepCols;
	private PreparedStatement prepRowkey;

	public ColHis findCols(String db, String tb, long time) {
		String key = String.format("%s|%s", db, tb).toLowerCase();
		java.sql.Connection conn = null;
		try {
			String url = String.format("jdbc:mysql://%s:%s?autoReconnect=true&useUnicode=true&characterEncoding=utf-8",
					connConf.getIp(), connConf.getPort());
			conn = JdbcConnection.getConnection("com.mysql.jdbc.Driver", url, connConf.getUser(),
					connConf.getPassword());
			if (prepCols == null || prepCols.isClosed()) {//
				prepCols = conn.prepareStatement(
						"select   column_name,data_type   from  information_schema.columns  where  table_schema=? and table_name=?");
			}
			List<String> ret = new ArrayList<>();
			List<String> retType = new ArrayList<>();
			JdbcAssit.setPreParam(prepCols, db, tb);
			ResultSet rs = prepCols.executeQuery();
			while (rs.next()) {
				ret.add(rs.getString(1));
				retType.add(rs.getString(2));
			}
			rs.close();
			if (CollectionUtils.isEmpty(ret)) {
				String msg = String.format("db:%s,td:%s,user:%s 没有s查询到列名，请检查用户是否有此权限", db, tb, connConf.getUser());
				log.error(msg);
				throw new RuntimeException(msg);
				// close();
				// LoggerUtil.exit(JvmStatus.s15);
			}
			if (connConf.getRds()) {// 是rds
				if (prepRowkey == null || prepRowkey.isClosed()) {
					prepRowkey = conn.prepareStatement(
							"SELECT k.column_name FROM information_schema.table_constraints t JOIN information_schema.key_column_usage k USING (constraint_name,table_schema,table_name) WHERE t.constraint_type='PRIMARY KEY' AND t.table_schema=? AND t.table_name=?");

				}
				JdbcAssit.setPreParam(prepRowkey, db, tb);
				ResultSet rs2 = prepRowkey.executeQuery();
				if (!rs2.next()) {
					ret.add("_rowkey_");
					retType.add("varchar");
				}
				rs2.close();
			}
			Builder ColHisBuilder = ColHis.newBuilder();
			ColHisBuilder.setTime(time);
			ColHisBuilder.setTimeStr(DateFormatCase.YYYY_MM_DD_hhmmss.getInstanc()
					.format(time == -1 ? System.currentTimeMillis() : time * 1000));
			ColHisBuilder.setServerIp(connConf.getIp());
			ColHisBuilder.setDb(db);
			ColHisBuilder.setTb(tb);
			ColHisBuilder.addAllCols(ret);
			ColHisBuilder.addAllColTypes(retType);
			ColHis retobj = ColHisBuilder.build();
			// 保存
			saveCheckPoint.saveColName(retobj);
			// 放内存
			List<ColHis> findColsList = saveCheckPoint.findColsList(db, tb);
			// 预防没有排序
			Collections.sort(findColsList, new Comparator<ColHis>() {
				@Override
				public int compare(ColHis o1, ColHis o2) {
					long def = o2.getTime() - o1.getTime();
					return def > 0 ? 1 : (def < 0 ? -1 : 0);
				}
			});
			colsMap.put(key, findColsList);
			return retobj;
		} catch (Exception e) {
			log.error("获取cols错误", e);
			throw new RuntimeException("获取cols错误");
		} finally {
			if (conn != null) {
				try {
					conn.close();
				} catch (SQLException e) {
					log.error("关闭连接失败", e);
				}
			}
		}
	}

	/***
	 * 验证col是否正确，如果map没有代表是新的，需要读数据库获得
	 * 
	 * @param db
	 * @param tb
	 * @param time
	 * @return
	 */
	public ColHis ValidKey(String db, String tb, long time) {
		String key = String.format("%s|%s", db, tb).toLowerCase();
		if (colsMap.containsKey(key)) {
			List<ColHis> list = colsMap.get(key);
			ColHis selFitColName = selFitColName(time, list);
			if (selFitColName == null) {
				throw new RuntimeException("没有可用的col信息。");// 由于第1条时间设置为-1,所以它一般不会被执行
			}
			return selFitColName;
		} else {
			List<ColHis> findColsList = saveCheckPoint.findColsList(db, tb);// 需要排好序
			if (CollectionUtils.isEmpty(findColsList)) {
				ColHis retobj = findCols(db, tb, -1);// 没有列名就去查一次
				List<ColHis> retlist = new ArrayList<>();
				retlist.add(retobj);
				colsMap.put(key, retlist);
				return retobj;
			} else {
				// 预防没有排序
				Collections.sort(findColsList, new Comparator<ColHis>() {
					@Override
					public int compare(ColHis o1, ColHis o2) {
						long def = o2.getTime() - o1.getTime();
						return def > 0 ? 1 : (def < 0 ? -1 : 0);
					}
				});
				colsMap.put(key, findColsList);
				ColHis selFitColName = selFitColName(time, findColsList);
				if (selFitColName == null) {
					throw new RuntimeException("没有可用的col信息。");// 由于第1条时间设置为-1,所以它一般不会被执行
				}
				return selFitColName;
			}

		}
	}

	private ColHis selFitColName(long time, List<ColHis> list) {
		for (int i = 0; i < list.size(); i++) {
			ColHis colHis = list.get(i);
			if (colHis.getTime() < time) {
				return colHis;
			}
		}
		return null;
	}

	private void initData(DuckulaEvent.Builder rowbuilder, List<String[]> datas, List<String> cols, int rowNo) {
		if (CollectionUtils.isEmpty(datas)) {
			return;
		}
		String[] tempary = datas.get(rowNo);
		for (int i = 0; i < tempary.length; i++) {
			if (tempary[i] != null) {
				rowbuilder.putAfter(cols.get(i), tempary[i]);
			}
		}
	}
}
